/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.genitus.drum.service

import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.{Executors, TimeUnit}

import com.google.common.util.concurrent.{AbstractIdleService, ThreadFactoryBuilder}
import com.google.inject.Inject
import com.lmax.disruptor.RingBuffer._
import com.lmax.disruptor.util.DaemonThreadFactory
import com.lmax.disruptor._
import org.genitus.drum.conf.{BuffConf, BuffType}
import org.genitus.drum.core.{Handler, MessageEvent}
import org.genitus.drum.model.StatData
import org.slf4j.LoggerFactory

/**
  * Message buffer service.
  *
  * @author gwjiang (gwjiang@iflytek.com), 2016/8/22.
  */
class BuffService @Inject() (
  val conf: BuffConf,
  val handlers: Array[Handler]
) extends AbstractIdleService {
  /** logger. */
  private[this] val log = LoggerFactory.getLogger(getClass.getName)

  /** starat timestamp. */
  private[this] final val startTime = System.currentTimeMillis()

  /** wait strategy */
  private[this] val waitStrategy: WaitStrategy = conf.buffWaitStrategy() match {
    case BuffType.BLOCKING =>
      log.info(s"RMQ sink wait strategy use: ${BuffType.BLOCKING}")
      new BlockingWaitStrategy
    case BuffType.BUSYSPIN =>
      log.info(s"RMQ sink wait strategy use: ${BuffType.BUSYSPIN}")
      new BusySpinWaitStrategy
    case BuffType.YIELDING =>
      log.info(s"RMQ sink wait strategy use: ${BuffType.YIELDING}")
      new YieldingWaitStrategy
    case BuffType.SLEEPING =>
      log.info(s"RMQ sink wait strategy use: ${BuffType.SLEEPING}")
      new SleepingWaitStrategy
    case _ =>
      log.info(s"RMQ sink wait strategy use: ${BuffType.SLEEPING}")
      new SleepingWaitStrategy
  }

  /** RB. */
  private[this] val ringBuffer = createMultiProducer(
    MessageEvent.EVENT_FACTORY, conf.buffSize(), waitStrategy)

  /** worker pool. */
  private[this] var workerPool: WorkerPool[MessageEvent] = _

  /** scheduled drainer and reporter thread pool. */
  private[this] val timedDrainerPool = Executors.newScheduledThreadPool(1,
    new ThreadFactoryBuilder().setNameFormat("app-drainer/reporter-%d").setDaemon(true).build)

  /** msg accepted num. */
  private[this] val accepted = new AtomicLong(0)

  /** last msg accepted num. */
  private[this] var lastAccepted = 0L

  /** last takes num. */
  private[this] val handlerLastTakes = Array.fill[Long](handlers.length) { 0L }

  /**
    * Append message to buffer.
    *
    * @param key message key.
    * @param body message log.
    */
  def append(key: String, body: Array[Byte]): Unit = {
    val sequence: Long = ringBuffer.next
    try {
      val event: MessageEvent = ringBuffer.get(sequence)
      event.key = key
      event.body = body
    } finally {
      ringBuffer.publish(sequence)
      accepted.incrementAndGet()
    }
  }

  /**
    * Append messages (key, bytes) to buffer in batch.
    *
    * @param messages messages.
    */
  def append(messages: List[(String, Array[Byte])]): Unit = {
    if (messages.size == 1) {
      append(messages.head._1, messages.head._2)
    } else if (messages.size > 1) {
      val hi = ringBuffer.next(messages.size)
      val lo = hi - (messages.size - 1)
      var l = lo
      var i = 0

      try {
        while (l <= hi) {
          val event = ringBuffer.get(l)
          event.key = messages(i)._1
          event.body = messages(i)._2

          l += 1
          i += 1
        }
      } finally {
        ringBuffer.publish(lo, hi)
        accepted.addAndGet(messages.size)
      }
    }
  }

  /**
    * @inheritdoc
    */
  override def startUp(): Unit = {
    log.debug("start BuffService ...")

    val executor = Executors.newFixedThreadPool(
      conf.buffWokers(), DaemonThreadFactory.INSTANCE)
    workerPool = new WorkerPool[MessageEvent](
      ringBuffer, ringBuffer.newBarrier(), new FatalExceptionHandler(), handlers: _*)
    // set gating and start
    ringBuffer.addGatingSequences(workerPool.getWorkerSequences: _*)
    workerPool.start(executor)

    // start drain thread
    timedDrainerPool.scheduleAtFixedRate(
      new Runnable {
        override def run(): Unit = handlers.foreach(_.drain())
      }, 10, 10, TimeUnit.SECONDS
    )

    // start report thread
    timedDrainerPool.scheduleAtFixedRate(
      new Runnable {
        override def run(): Unit = {
          val num = accepted.get()
          log.info("source.accepted = " + num
            + ", tps = " + "%,d ops/sec".format((num - lastAccepted) / 10))
          lastAccepted = num

          var id = 0
          handlers.foreach({ x =>
            log.info("sinker[" + id + "].puts = "
              + x.reportProgress()._1 + ", sinker[" + id + "].takes = "
              + x.reportProgress()._2 + ", tps = " + "%,d ops/sec".format(
              (x.reportProgress()._2 - handlerLastTakes(id)) / 10))
            handlerLastTakes(id) = x.reportProgress()._2
            id += 1
          })
        }
      },
      10,
      10,
      TimeUnit.SECONDS)
  }

  override def shutDown(): Unit = {
    log.debug("stop BuffService ...")

    if (workerPool != null) {
      workerPool.drainAndHalt()
    }
    handlers.foreach(x => x.close())

    // shutdown timedDrainerPool
    timedDrainerPool.shutdown()
    try {
      while (!timedDrainerPool.isTerminated) {
        timedDrainerPool.awaitTermination(1000, TimeUnit.MILLISECONDS)
      }
    } catch {
      case ex: InterruptedException =>
        log.warn("shutdown interrupted on " + timedDrainerPool, ex)
    }
  }

  /**
    * Gets stat data.
    *
    * @return stat data.
    */
  def getStat: StatData = {
    val (putCount, takeCount) = handlers.foldLeft((0L, 0L))({ (x, item) =>
      (x._1 + item.reportProgress()._1, x._2 + item.reportProgress()._2)
    })
    val acceptedCount = this.accepted.get()

    StatData(
      startTime = this.startTime,
      accepted = acceptedCount,
      puts = putCount,
      takes = takeCount,
      tps = (acceptedCount - lastAccepted) / 10
    )
  }
}
